package com.hung.triple.heartbeat;

import com.hung.triple.consumer.response.FuturePool;
import com.hung.triple.consumer.response.TripleFuture;
import com.hung.triple.protocol.TripleProtocol;
import com.hung.triple.protocol.rpc.TripleRpcRequest;
import com.hung.triple.protocol.rpc.TripleRpcResponse;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.Timeout;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * 参考资料:
 * https://blog.csdn.net/beyondself_77/article/details/81026251
 *
 * @author Hung
 * @date 2023/2/7 23:01
 */
@Slf4j
public class RpcHeartbeatTrigger implements HeartbeatTrigger {

    public static final Integer maxCount = 3;
    private static final long heartbeatTimeoutMillis = 1000L;
    private CommandFactory commandFactory;

    public RpcHeartbeatTrigger(CommandFactory commandFactory) {
        this.commandFactory = commandFactory;
    }

    public void heartbeatTriggered(final ChannelHandlerContext ctx) {
        //这里是Client/Consumer  专属的心跳处理器
        //Channel中有专属的 HEARTBEAT_COUNT 属性
        //如果发送心跳次数

        Integer heartbeatTimes = ctx.channel().attr(Connection.HEARTBEAT_COUNT).get();
        final Connection conn = ctx.channel().attr(Connection.CONNECTION).get();

        if (heartbeatTimes >= maxCount) {
            //心跳次数大于最大值 就关闭channel
            try {
                conn.close();
                log.error("Heartbeat failed for {} times, close the connection from client side: {} ", heartbeatTimes, RemotingUtil.parseRemoteAddress(ctx.channel()));
            } catch (Exception var8) {
                log.warn("Exception caught when closing connection in SharableHandler.", var8);
            }
        } else {
            //如果心跳开关为false就代表不需要心跳 直接跳过
            boolean heartbeatSwitch = ctx.channel().attr(Connection.HEARTBEAT_SWITCH).get();
            if (!heartbeatSwitch) {
                return;
            }

            //TODO 问题3
            TripleRpcRequest heartBeat = new TripleRpcRequest();
            heartBeat.setType(0);
            TripleFuture tripleFuture = FuturePool.push(heartBeat);

            tripleFuture.addListeners((success, result, throwable) -> {
                TripleRpcResponse response;
                try {
                    response = tripleFuture.get();
                } catch (InterruptedException var4) {
                    RpcHeartbeatTrigger.log.error("Heartbeat ack process error! Id={}, from remoteAddr={}", new Object[]{heartBeat.getRequestId(), RemotingUtil.parseRemoteAddress(ctx.channel()), var4});
                    return;
                } catch (ExecutionException e) {
                    throw new RuntimeException(e);
                }

                if (response != null && response.getCode() == TripleRpcResponse.SUCCESS) {
                    log.debug("Heartbeat ack received! Id={}, from remoteAddr={}", response.getRequestId(), RemotingUtil.parseRemoteAddress(ctx.channel()));

                    ctx.channel().attr(Connection.HEARTBEAT_COUNT).set(0);
                } else {
                    if (response == null) {
                        log.error("Heartbeat timeout! The address is {}", RemotingUtil.parseRemoteAddress(ctx.channel()));
                    } else {
                        log.error("Heartbeat exception caught! Error code={}, The address is {}", response.getCode(), RemotingUtil.parseRemoteAddress(ctx.channel()));
                    }

                    Integer times = ctx.channel().attr(Connection.HEARTBEAT_COUNT).get();
                    ctx.channel().attr(Connection.HEARTBEAT_COUNT).set(times + 1);
                }
            });

            int heartbeatId = heartBeat.getRequestId();
            conn.addInvokeFuture(tripleFuture);

            log.debug("Send heartbeat, successive count={}, Id={}, to remoteAddr={}", heartbeatTimes, heartbeatId, RemotingUtil.parseRemoteAddress(ctx.channel()));

            //一秒没有成功的话就执行future
            Timeout timeout = TimerHolder.getTimer().newTimeout(timeout1 -> {
                TripleFuture future = conn.removeInvokeFuture(heartbeatId);
                if (future != null && !future.isHaveCallBack() && future.isDone()) {
                    future.finishedInvoke(commandFactory.createSuccessResponse(conn.getRemoteAddress()));
                } else {
                    //失败的话执行回调
                    future.finishedInvoke(commandFactory.createTimeoutResponse(conn.getRemoteAddress()));
                }
            }, 1000L, TimeUnit.MILLISECONDS);

            tripleFuture.addTimeout(timeout);

            ctx.writeAndFlush(TripleProtocol.buildRpcRequest(heartBeat)).addListener((ChannelFutureListener) channelFuture -> {
                if (channelFuture.isSuccess()) {
                    log.debug("Send heartbeat done! Id={}, to remoteAddr={}", heartbeatId, RemotingUtil.parseRemoteAddress(ctx.channel()));

                } else {
                    TripleFuture future = conn.removeInvokeFuture(heartbeatId);
                    if (future != null) {
                        future.cancelTimeout();
                        //TODO 发送FailedResponse
                    }
                    RpcHeartbeatTrigger.log.error("Send heartbeat failed! Id={}, to remoteAddr={}", heartbeatId, RemotingUtil.parseRemoteAddress(ctx.channel()));
                }

            });


        }

    }
}
